package com.xrui.adc.hbase.schema;

import com.google.common.base.Preconditions;

import com.xrui.adc.hbase.schema.BinaryTableSchemaResolver;
import com.xrui.adc.hbase.codec.RowKeyCodec;
import com.xrui.adc.hbase.hashing.BinaryRowKeyHasher;
import com.xrui.hbase.TableName;
import com.xrui.hbase.conversion.AvroCellConverter;
import com.xrui.hbase.exception.InvalidLayoutException;
import com.xrui.hbase.impl.DefaultHTableInterfaceFactory;
import com.xrui.hbase.schema.*;
import com.xrui.hbase.schema.desc.CellSchemaDesc;
import com.xrui.hbase.schema.desc.ColumnDesc;
import com.xrui.hbase.schema.desc.TableLayoutDesc;
import org.apache.avro.Schema;
import org.apache.hadoop.conf.Configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public final class BinaryTableLayoutResolver implements TableLayoutResolver {
    private static final Logger LOG = LoggerFactory.getLogger(BinaryTableLayoutResolver.class);

    private BinaryTableSchemaResolver mSchemaTableResolver;

    /**
     * {@inheritDoc}
     */
    @Override
    public synchronized TableLayout resolve(String tableName, Object resource) throws InvalidLayoutException {
        TableName table = TableName.valueOf(tableName);
        TableName schemaTable = TableName.TableNameBuilder.newBuilder()
            .withNamespace(table.getNamespace())
            .withPrefix(table.getPrefix())
            .withSeparator(table.getSeparator())
            .withIdentifier(BinaryTableSchemaResolver.SCHEMA_TABLE_NAME)
            .build();
        if (null == mSchemaTableResolver) {
            try {
                mSchemaTableResolver = new BinaryTableSchemaResolver(
                    schemaTable,
                    (Configuration) resource,
                    DefaultHTableInterfaceFactory.get()
                );
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }

        Schema schema = mSchemaTableResolver.apply(table.getIdentifier());
        Preconditions.checkNotNull(schema);

        List<RowKeyComponent> rowKeyComponents = new ArrayList<>();
        rowKeyComponents.add(RowKeyComponent.create("key", RowKeyComponent.ComponentType.STRING, true));

        RowKeyFormat rowKeyFormat = RowKeyFormat.RowKeyFormatBuilder.newBuilder()
            .withDecoder(RowKeyCodec.PROVIDER_NAME)
            .withDelimiter("|")
            .withHasher(BinaryRowKeyHasher.PROVIDER_NAME)
            .withEncoding(RowKeyEncoding.FORMATTED)
            .withHashSize(8)
            .withEncoder(RowKeyCodec.PROVIDER_NAME)
            .withComponents(rowKeyComponents)
            .build();
        TableLayoutDesc tableLayoutDesc = TableLayoutDesc.TableLayoutBuilder.newBuilder()
            .withTableName(tableName)
            .withRowKey(rowKeyFormat)
            .addColumn(
                ColumnDesc.ColumnBuilder.newBuilder()
                    .withFamily("cf")
                    .withCellSchema(CellSchemaDesc.CellSchemaBuilder.newBuilder()
                        .withType(SchemaType.AVRO)
                        .withID(table.getIdentifier())
                        .withSchema(schema.toString())
                        .withConverter(AvroCellConverter.PROVIDER_NAME)
                        .build()
                    )
                    .build()
            ).build();
        return TableLayout.newLayout(tableLayoutDesc).setSchemaResolver(mSchemaTableResolver);
    }

    @Override
    public String getName() {
        return this.getClass().getName();
    }
}
